package com.roy.netty.client;

import java.net.InetSocketAddress;

import io.netty.channel.Channel;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.channel.pool.ChannelPoolMap;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.util.concurrent.Future;

public class PooledNettyTCPClient extends AbstractTCPClient {

	private ChannelPoolMap<InetSocketAddress, FixedChannelPool> poolMap;
	
	public PooledNettyTCPClient() {
        createChannelPoolMap();
	}
	private void createChannelPoolMap() {
		this.poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() {
            @Override
            protected FixedChannelPool newPool(InetSocketAddress key) {
                return new FixedChannelPool(getBootstrap().remoteAddress(key),
                        new ClientChannelPoolHandler(getChannelInitializer()),
                        ChannelHealthChecker.ACTIVE, FixedChannelPool.AcquireTimeoutAction.FAIL,
                        getSettings().getTransportPoolAcquireTimeoutMillis(),
                        getSettings().getTransportPoolMaxConnections(),
                        getSettings().getTransportPoolMaxPendingAcquires());
            }
        };
	}
	@Override
	public String doSend(InetSocketAddress address, String message,
			int waitMilliSeconds) throws Throwable {
		FixedChannelPool pool = poolMap.get(address);
        Future<Channel> channelFuture = pool.acquire().sync();
        if (channelFuture.isSuccess()) {
            Channel ch = channelFuture.getNow();
            try {
                MessageSender sender = getMessageSender(ch);
                return sender.sendMessage(message, waitMilliSeconds);
            } finally {
                pool.release(ch);
            }
        } else {
            System.out.println("Connect server " + address + " error ");
            throw channelFuture.cause();
        }
	}

}
